package com.sc.passbook.service.impl;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sc.passbook.constant.Constant;
import com.sc.passbook.dao.MerchantsDao;
import com.sc.passbook.entity.Merchants;
import com.sc.passbook.mapper.PassTemplateMapper;
import com.sc.passbook.service.IInventoryService;
import com.sc.passbook.service.IUserPassService;
import com.sc.passbook.utils.RowKeyGen;
import com.sc.passbook.vo.*;
import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.LongComparator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.ParseException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toList;

/**
 * @author sc
 * Created on  2018/9/2
 */
@Service
@Slf4j
public class InventoryService implements IInventoryService {

    @Autowired
    private HbaseTemplate hbaseTemplate;
    @Autowired
    private MerchantsDao merchantsDao;
    @Autowired
    private IUserPassService userPassService;

    @Override
    public Response getInventoryInfo(Long userId) throws IOException, ParseException {
        Response userAllPassInfo = userPassService.getUserAllPassInfo(userId);
        List<PassInfo> list = (List<PassInfo>) userAllPassInfo.getData();
        List<PassTemplate> ex = list.stream().map(PassInfo::getPassTemplate).collect(Collectors.toList());
        List<String> exIds = Lists.newArrayList();
        ex.forEach(e->exIds.add(RowKeyGen.genPassTemplateRowKey(e)));
        return Response.success(new InventoryResponse(userId,buildPassTemplateInfo(getAvaliablePT(exIds))));
    }

    private List<PassTemplateInfo> buildPassTemplateInfo(List<PassTemplate> passTemplates){
        Map<Integer,Merchants> maps = Maps.newHashMap();
        List<Integer> ids = passTemplates.stream().map(PassTemplate::getId).collect(toList());
        List<Merchants> merchants = merchantsDao.findByIdIn(ids);
        merchants.forEach(m->maps.put(m.getId(),m));
        List<PassTemplateInfo> res = Lists.newArrayListWithExpectedSize(passTemplates.size());
        for (PassTemplate passTemplate : passTemplates) {
            Merchants mc = maps.getOrDefault(passTemplate.getId(), null);
            if(mc==null){
               log.error("找不到商户");
               continue;
            }
            res.add(new PassTemplateInfo(mc,passTemplate));
        }
        return res;
    }

    private List<PassTemplate> getAvaliablePT(List<String> eIds){
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ONE);
        filterList.addFilter(new SingleColumnValueFilter(
                Bytes.toBytes(Constant.PassTemplateTable.FAMILY_C),
                Bytes.toBytes(Constant.PassTemplateTable.LIMIT),                      CompareFilter.CompareOp.GREATER,
                new LongComparator(0L)
        ));
        filterList.addFilter(new SingleColumnValueFilter(
                Bytes.toBytes(Constant.PassTemplateTable.FAMILY_C),
                Bytes.toBytes(Constant.PassTemplateTable.LIMIT),                      CompareFilter.CompareOp.EQUAL,
                Bytes.toBytes("-1")
        ));

        Scan scan = new Scan();
        scan.setFilter(filterList);
        List<PassTemplate> passTemplates = hbaseTemplate.find(Constant.PassTemplateTable.TABLE_NAME,
                scan, new PassTemplateMapper());
        List<PassTemplate> res = Lists.newArrayList();
        Date curr = new Date();
        for (PassTemplate passTemplate : passTemplates) {
            if(eIds.contains(RowKeyGen.genPassTemplateRowKey(passTemplate))){
                continue;
            }
            if(curr.getTime()>=passTemplate.getStart().getTime()&&
                    curr.getTime()<=passTemplate.getEnd().getTime()){
                res.add(passTemplate);
            }
        }
        return res;
    }
}
